package com.xujl.task;


import java.util.Set;
import java.util.WeakHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 任务管理类
 * 线程池管理
 */
public class RxExecutor {
    private static final String TAG = "RxExecutor";
    private ExecutorService mExecutor;
    private WeakHashMap<Task, String> mWeakTaskMap = new WeakHashMap<>();
    private WeakHashMap<TaskCallable, String> mWeakThreadMap = new WeakHashMap<>();

    private RxExecutor () {
        super();
    }

    /**
     * 使用默认参数初始化线程池
     * 核心线程数为cpu核心数
     */
    public void init () {
        //cpu核心数
        final int processors = Runtime.getRuntime().availableProcessors();
        init(processors * 4);
    }

    /**
     * @param corePoolSize 线程池中核心子线程数量，通常情况下建议不超过cpu核心数
     *                     如果应用需要高并发，建议不要超cpu核心数*4,需要注意的是，如果
     *                     线程达到线程池上限，且每个线程均没有退出，则后续线程无法
     *                     执行
     */
    public void init (int corePoolSize) {
        /*参数从左到右依次为，1：核心线程数量:2：最大线程数量（超出部分会进行等待），
            3：线程空闲超时时间（超时后将被关闭），4:超出的线程队列

         */
        mExecutor = new ThreadPoolExecutor(corePoolSize, corePoolSize * 3,
                2, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(1024)
                , new ThreadPoolExecutor.AbortPolicy());
        ((ThreadPoolExecutor) mExecutor).prestartAllCoreThreads(); // 预启动所有核心线程
    }

    /**
     * 完全自定义进行初始化
     *
     * @param executor
     */
    public void init (ExecutorService executor) {
        mExecutor = executor;
    }

    public <T> void executeTask (Task<T> task) {
        if (task.getTaskState().equals(Task.STATE_RUNNING)) {
            RxLog.i(TAG, task.getTaskId() + "：该任务正在执行中，请勿重复执行！");
            return;
        }
        final TaskCallable<T> thread = new TaskCallable<>();
        mWeakThreadMap.put(thread, task.getTaskId());
        executeTask(thread, task);
    }

    /**
     * @param thread 子线程任务
     * @param task   任务关联对象
     */
    public <T> void executeTask (TaskCallable<T> thread, Task<T> task) {
        task.reset();
        thread.setTask(task);
        if (mExecutor == null) {
            init();
        }
        if (mExecutor.isShutdown()) {
            RxLog.d(TAG, "线程池已shutdown,无法添加新任务！！！");
            return;
        }
        final Future<T> submit = mExecutor.submit(thread);
        task.setFuture(submit);
        mWeakTaskMap.put(task, task.getTaskId());
    }


    /**
     * 仅仅运行一个主线程上的任务，
     *
     * @param task
     */
    public void executeUiTask (Task<?> task) {
        final Emitter<?> emitter = task.getEmitter();
        if (emitter != null) {
            emitter.runUi();
        }
    }

    /**
     * 执行一个单纯的线程
     *
     * @param runnable
     */
    public void executeRunnableTask (Runnable runnable) {
        if (mExecutor == null) {
            init();
        }
        if (mExecutor.isShutdown()) {
            RxLog.d(TAG, "线程池已shutdown,无法添加新任务！！！");
            return;
        }
        if (runnable != null) {
            mExecutor.execute(runnable);
        }
    }

    /**
     * 通过id取消任务
     *
     * @param taskId
     */
    public boolean cancelTaskForId (String taskId, boolean interrupt) {
        final Set<Task> keySet = mWeakTaskMap.keySet();
        for (Task task : keySet) {
            if (task != null && taskId.equals(mWeakTaskMap.get(task))) {
                return task.cancel(interrupt);
            }
        }
        return false;
    }

    /**
     * 关闭线程池
     *
     * @param isInterrupt 是否中断正在执行的任务
     */
    public void shutDown (boolean isInterrupt) {
        mExecutor.shutdown();
        if (isInterrupt) {
            try {
                // Wait a while for existing tasks to terminate
                if (!mExecutor.awaitTermination(2, TimeUnit.SECONDS)) {
                    mExecutor.shutdownNow(); // Cancel currently executing tasks
                }
            } catch (InterruptedException ie) {
                // (Re-)Cancel if current thread also interrupted
                mExecutor.shutdownNow();
                // Preserve interrupt status
                Thread.currentThread().interrupt();
            }

        }
    }

    public void shutDown () {
        shutDown(false);
    }

    public void setDebug (boolean isDebug) {
        RxLog.sIsDebug = isDebug;
    }

    /**
     * 打印当前线程池线程数量状态
     */
    public void printRunnableMsg () {
        ThreadPoolExecutor executor = (ThreadPoolExecutor) mExecutor;
        RxLog.i(TAG,
                "TaskQueueCount:" + executor.getQueue().size() +
                        "ActiveCount:" + executor.getActiveCount()
        );
    }

    /**
     * 打印任务集合中任务的存活状态
     */
    public void printWeakMapMsg () {
        RxLog.i(TAG, "当前尚未回收任务数量：" + mWeakTaskMap.size());
        final Set<Task> keySet = mWeakTaskMap.keySet();
        for (Task task : keySet) {
            if (task != null) {
                RxLog.i(TAG, "当前尚未回收任务：" + mWeakTaskMap.get(task));
            }
        }
        RxLog.i(TAG, "当前尚未回收线程数量：" + mWeakThreadMap.size());
        final Set<TaskCallable> keys = mWeakThreadMap.keySet();
        for (TaskCallable callable : keys) {
            if (callable != null) {
                RxLog.i(TAG, "当前尚未回收线程：" + mWeakThreadMap.get(callable));
            }
        }
    }

    public ExecutorService getExecutor () {
        return mExecutor;
    }

    public static RxExecutor getInstance () {
        return Holder.RX_EXECUTOR;
    }


    private static class Holder {
        private static final RxExecutor RX_EXECUTOR = new RxExecutor();
    }

}
